package com.jili.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * ClassName: FlowDemo
 * Package com.jili.flow
 * Description:
 *
 * @Author: 廖继礼
 * @Create: 2025/3/22
 * @Version: v1.0
 */
public class FlowDemo {

    // 中间处理器
    static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("Processor订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            System.out.println("Processor拿到数据");
            item += "：你好";
            subscription.request(1);
            submit(item);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 1.定义一个发布者，发布数据
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        MyProcessor myProcessor = new MyProcessor();

        // 2.定义一个订阅者，订阅者感兴趣发布者的数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始了：" + subscription);
                this.subscription = subscription;
                // 从上游请求数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "订阅者，接收到数据" + item);
                if ("p-7".equals(item)) {
                    subscription.cancel();
                } else {
                    subscription.request(1);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "订阅者，接收到错误信息" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread() + "订阅者接收到完成信号");
            }
        };
        publisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            // 发布10条数据
            publisher.submit("p-" + i);
            // publisher发布的所有数据在buffer区
        }
        publisher.close();
        Thread.sleep(1000);
    }
}
